package cn.ccydrools;

import cn.ccydrools.pojo.DroolsRulesBean;
import cn.ccydrools.pojo.a03Event;
import cn.ccydrools.pojo.RulesStateBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.kie.api.io.ResourceType;
import org.kie.api.runtime.KieSession;
import org.kie.internal.utils.KieHelper;

import java.util.Map;

public class _05FlinkDroolsDemo2 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        String servers = "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092";
        FlinkKafkaConsumer<String> rulesSource = _06KafkaSourceUtil.getKafkaSource(servers, "earliest", "drools-rules19");
        FlinkKafkaConsumer<String> eventsSource = _06KafkaSourceUtil.getKafkaSource(servers, "earliest", "event19");

        DataStreamSource<String> rulesStream = env.addSource(rulesSource);

        DataStreamSource<String> eventsStream = env.addSource(eventsSource);

        SingleOutputStreamOperator<DroolsRulesBean> rulesBeanStream = rulesStream.process(new ProcessFunction<String, DroolsRulesBean>() {

            @Override
            public void processElement(String line, Context ctx, Collector<DroolsRulesBean> out) throws Exception {

                try {
                    JSONObject jsonObject = JSON.parseObject(line);
                    String type = jsonObject.getString("type");
                    if ("INSERT".equals(type) || "UPDATE".equals(type)) {
                        JSONArray jsonArray = jsonObject.getJSONArray("data");
                        for (int i = 0; i < jsonArray.size(); i++) {
                            DroolsRulesBean bean = jsonArray.getObject(i, DroolsRulesBean.class);
                            out.collect(bean);
                        }
                    }
                } catch (Exception e) {
                    //e.printStackTrace();
                }
            }
        });


        MapStateDescriptor<Integer, RulesStateBean> stateDescriptor = new MapStateDescriptor<>("rule-state", Integer.class, RulesStateBean.class);

        BroadcastStream<DroolsRulesBean> broadcastStream = rulesBeanStream.broadcast(stateDescriptor);

        ////u1001,c201,view
        SingleOutputStreamOperator<Tuple3<String, String, String>> tpStream = eventsStream.map(new MapFunction<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(fields[0], fields[1], fields[2]);
            }
        });

        //进行KeyBy
        tpStream.keyBy(t -> t.f0).connect(broadcastStream).process(new KeyedBroadcastProcessFunction<String, Tuple3<String, String, String>, DroolsRulesBean, Tuple3<String, String, String>>() {

            private transient MapState<Tuple2<String, String>, Integer> mapState;


            @Override
            public void open(Configuration parameters) throws Exception {
                //定义一个状态描述器
                //((c201,view), 3)
                MapStateDescriptor<Tuple2<String, String>, Integer> stateDescriptor = new MapStateDescriptor<>("category-event-count", TypeInformation.of(new TypeHint<Tuple2<String, String>>() {
                }), TypeInformation.of(new TypeHint<Integer>() {
                }));

                mapState = getRuntimeContext().getMapState(stateDescriptor);
            }

            @Override
            public void processElement(Tuple3<String, String, String> value, ReadOnlyContext ctx, Collector<Tuple3<String, String, String>> out) throws Exception {

                String uid = value.f0; //u1001
                String cid = value.f1; //c201
                String type = value.f2; //view

                Tuple2<String, String> key = Tuple2.of(cid, type);
                Integer count = mapState.get(key);
                if (count == null) {
                    count = 0;
                }
                count++;
                mapState.put(key, count);

                a03Event event = new a03Event(type, count, false);

                Iterable<Map.Entry<Integer, RulesStateBean>> entries = ctx.getBroadcastState(stateDescriptor).immutableEntries();

                for (Map.Entry<Integer, RulesStateBean> entry : entries) {
                    Integer rulesId = entry.getKey();
                    //KieSession
                    //SQL
                    //Params
                    RulesStateBean rulesStateBean = entry.getValue();

                    KieSession kieSession = rulesStateBean.getKieSession();
                    kieSession.insert(event);
                    //应用所有的规则
                    kieSession.fireAllRules();

                    if (event.isHit()) {
                        out.collect(Tuple3.of(uid, "发优惠卷", "满30减5"));
                    }
                }
            }

            @Override
            public void processBroadcastElement(DroolsRulesBean bean, Context ctx, Collector<Tuple3<String, String, String>> out) throws Exception {

                BroadcastState<Integer, RulesStateBean> broadcastState = ctx.getBroadcastState(stateDescriptor);

                short status = bean.getStatus();
                Integer rulesId = bean.getId();
                if (status == 1 || status == 2) {
                    //添加或更新
                    KieHelper kieHelper = new KieHelper();
                    kieHelper.addContent(bean.getCode(), ResourceType.DRL);
                    KieSession kieSession = kieHelper.build().newKieSession();
                    RulesStateBean stateBean = new RulesStateBean(rulesId, kieSession, null, bean.getStart_time(), bean.getEnd_time(), bean.getCounts());
                    broadcastState.put(rulesId, stateBean);
                } else if (status == 3) {
                    //删除
                    broadcastState.remove(rulesId);
                }
            }
        }).print();


        env.execute();
    }
}
